package net.bwie.realtime.gongdan.dwd.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * 用户行为数据
 * @Author: FuHe
 * @Date: 2025/6/10
 * todo 大数据-用户画像-12-达摩盘家庭特征
 */
public class GongdanUserLogJob {
    public static void main(String[] args) {
        // 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 输入表  input 映射到kafka消息队列
        createInputTable(tabEnv);
        // 数据处理
        Table reportTable = handle(tabEnv);
        // 输出表  output  映射到clickhouse表
        createOutputTable(tabEnv);
        // 保存数据
        saveToDoris(tabEnv, reportTable);
    }

    private static void saveToDoris(TableEnvironment tabEnv, Table reportTable) {
        tabEnv.createTemporaryView("report_table", reportTable);
//        tabEnv.sqlQuery("select * from report_table").execute().print();
        tabEnv.executeSql("INSERT INTO doris_user_behavior_sink SELECT * FROM report_table");
    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE doris_user_behavior_sink\n" +
                "(\n" +
                "    log_id         STRING COMMENT '日志ID',\n" +
                "    user_id        STRING COMMENT '用户ID',\n" +
                "    item_id        STRING COMMENT '商品ID',\n" +
                "    behavior_type  STRING COMMENT '行为类型:purchase/browse/collect/add_cart/search',\n" +
                "    behavior_time  STRING COMMENT '行为时间',\n" +
                "    search_keyword STRING COMMENT '搜索关键词',\n" +
                "    session_id     STRING COMMENT '会话ID',\n" +
                "    stay_duration  BIGINT COMMENT '停留时长(ms)',\n" +
                "    page_url       STRING COMMENT '页面URL'\n" +
                ") WITH (\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'url' = 'jdbc:clickhouse://node103:8123/gongdan12',\n" +
                "    'table' = 'user_behavior',\n" +
                "    'username' = 'default',\n" +
                "    'password' = '',\n" +
                "    'format' = 'json'\n" +
                ")");
    }

    private static Table handle(TableEnvironment tabEnv) {
        Table table = tabEnv.sqlQuery("select * from user_behavior_log");
        return table;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE user_behavior_log\n" +
                "(\n" +
                "    log_id         STRING COMMENT '日志ID',\n" +
                "    user_id        STRING COMMENT '用户ID',\n" +
                "    item_id        STRING COMMENT '商品ID',\n" +
                "    behavior_type  STRING COMMENT '行为类型:purchase/browse/collect/add_cart/search',\n" +
                "    behavior_time  STRING COMMENT '行为时间',\n" +
                "    search_keyword STRING COMMENT '搜索关键词',\n" +
                "    session_id     STRING COMMENT '会话ID',\n" +
                "    stay_duration  BIGINT COMMENT '停留时长(ms)',\n" +
                "    page_url       STRING COMMENT '页面URL'\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'user_behavior_topic',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                "    'properties.group.id' = 'user_behavior_group',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
//                "    'properties.auto.offset.reset' = 'latest',\n" +
                "    'format' = 'json',\n" +
                "    'json.fail-on-missing-field' = 'false',\n" +
                "    'json.ignore-parse-errors' = 'true'\n" +
                ")");
    }

    private static TableEnvironment getTableEnv() {
        // 环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        // 时区设置
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        // 并发设置
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态保存时间
        configuration.setString("table.exec.state.ttl", "5s");
        // 返回值
        return tabEnv;
    }
}
